package com.sg.java.apps;

import com.sg.java.PropertiesUtil;
import com.sg.java.ResourcePath;
import com.sg.java.util.HBaseUtils;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.OutputStreamWriter;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.StringJoiner;

public class CountHBase {

    private static final Logger log = LoggerFactory.getLogger(CountHBase.class);

    public static void main(String[] args) {
        String tableName = args[0];
        String path = args[1];
        int limit = Integer.parseInt(args[2]);
        if (!tableName.contains("cqqkjdyjc:")) {
            tableName = "cqqkjdyjc:" + tableName;
        }
        try (Connection hBaseConn = HBaseUtils.getHBaseConn(PropertiesUtil.createPropertiesFromResource(ResourcePath.hbase_properties));
             CSVPrinter csvPrinter = new CSVPrinter(new OutputStreamWriter(Files.newOutputStream(Paths.get(path)), StandardCharsets.UTF_8), CSVFormat.newFormat(',').withQuote(null).withRecordSeparator("\r\n").withIgnoreEmptyLines(true))) {
            TableName tn = TableName.valueOf(tableName);
            HTable hTable = (HTable) hBaseConn.getTable(tn);
            long totalCount = 0;
            long tempCount;
            Result next = null;
            int round = 0;
            long totalTime_s = System.currentTimeMillis();
            do {
                long temp_s = System.currentTimeMillis();
                tempCount = 0;
                round++;
                Scan scan = new Scan();
                scan.setLimit(limit);
                scan.addFamily(Bytes.toBytes("info"));
                if (tableName.contains("vol")) {
                    scan.addFamily(Bytes.toBytes("U"));
                    scan.setFilter(new SingleColumnValueFilter(
                            Bytes.toBytes("info"), Bytes.toBytes("PHASE_FLAG"), CompareOperator.EQUAL, Bytes.toBytes("1")
                    ));
                } else if (tableName.contains("power")) {
                    scan.addFamily(Bytes.toBytes("P"));
                } else if (tableName.contains("cur")) {
                    scan.addFamily(Bytes.toBytes("I"));
                }
                if (next != null) {
                    log.info(tn + "\t下次分页数据起始行：{}", Bytes.toString(next.getRow()));
                    scan.withStartRow(next.getRow(), false);
                }
                ResultScanner scanner = hTable.getScanner(scan);
                for (Result result : scanner) {
                    tempCount++;
                    totalCount++;
                    next = result;
                    StringJoiner sj = new StringJoiner(",");
                    sj.add(Bytes.toString(result.getRow()));
                    for (Cell cell : result.listCells()) {
                        sj.add(Bytes.toString(CellUtil.cloneFamily(cell)) + ":" + Bytes.toString(CellUtil.cloneQualifier(cell)) + "=" + Bytes.toString(CellUtil.cloneValue(cell)));
                    }
                    csvPrinter.printRecord(sj.toString());
                }
                long temp_e = System.currentTimeMillis() - temp_s;
                log.info("此轮：{}，条数：{}，耗时：{}ms", round, tempCount, temp_e);
            } while (tempCount >= limit);
            long totalTime_e = System.currentTimeMillis() - totalTime_s;
            log.info("全部完成 总耗时:{}ms 总条数:{}", totalTime_e, totalCount);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
